package messagepasser;

/**
 * 
 * @author Rishit Shroff(rshroff)
 * @author Huimin Yang(huiminy)
 * 
 */

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;




public class MessagePasser {

	private String myName;
	private int myPort;
	private String configFile;
	ListenSocket listeneningServer;
	private List<TimeStampedMessage> senderQ = new ArrayList<TimeStampedMessage>();
	
	private List<TimeStampedMessage> hbQ = new ArrayList<TimeStampedMessage>();
	private Map<String, Integer> sequenceNumber = new HashMap<String, Integer>();
	private int seqNo = 0;
	// name of nodes
	private List<String> nodeNames = new ArrayList<String>();
	private Map<String, NodeInfo> nodeList = new HashMap<String, NodeInfo>();
	
	private Map<String, String> csList = new HashMap<String, String>();
	private Map<String, String> crList = new HashMap<String, String>();
	private Map<String, String> isList = new HashMap<String, String>();
	private Map<String, String> irList = new HashMap<String, String>();
	
	private List<Message> inputBuffer = new ArrayList<Message>();
	private List<Message> outputBuffer = new ArrayList<Message>();
	private List<Message> ddBuffer = new ArrayList<Message>();
	private List<Message> nwBuffer = new ArrayList<Message>();
	private List<Message> readyBuffer = new ArrayList<Message>();
	private boolean prevNoDeliver = false;
	private Map<String, Boolean> OkList = new HashMap<String, Boolean>();
	private List<Message> OkMessages = new ArrayList<Message>();
	
	private ClockService clock;
	
	private long delay = 10;
	private long period = 5000;
	private int wait = 5;
	private String state = "Free";
	private long lengthOfCriticalSection = 5000;
	private Set<String> receivedMsg = new HashSet<String>();
	
	public MessagePasser(String configFile, ClockService clock) {
		readConfig(configFile);
		this.configFile = configFile;
		
		myPort = nodeList.get(this.myName).getPort();
		listeneningServer = new ListenSocket (myPort, this);
		listeneningServer.start();
		
		this.clock = clock;
		this.clock.initialize(nodeNames, myName);
		
		for (String process : nodeNames) {
			sequenceNumber.put(process, new Integer(0));
			OkList.put(process, false);
		}
	}

	public int getSeqNo() {
		return seqNo;
	}

	public void setSeqNo(int seqNo) {
		this.seqNo = seqNo;
	}
	public ClockService getClock() {
		return clock;
	}

	public void setClock(ClockService clock) {
		this.clock = clock;
	}

	
	
	/**
	 * Read the configuration file, and extract information.
	 * Use the information to configure MessagePasser
	 * 
	 * @param fileName	the name of the configuration file, include directory if necessary
	 *  
	 */
	public void readConfig(String fileName) {
		BufferedReader br = null;
		try {
			FileReader fr = new FileReader(fileName);
            br = new BufferedReader(fr);
            
            boolean eof = false;
            while (!eof) {
            	String line = br.readLine();
            	if (line == null) {
                    eof = true;
            	} else {
            		if (line.startsWith("ME ")) {
            			readMyName(line);
            		} else if (line.startsWith("NAMES")) {
            			readNames(line);
            		} else if (line.startsWith("node")) {
            			readNodes(line);
            		} else if (line.startsWith("cs")) {
            			readProcess(csList, line);
            		} else if (line.startsWith("cr")) {
            			readProcess(crList, line);
            		} else if (line.startsWith("is")) {
            			readProcess(isList, line);
            		} else if (line.startsWith("ir")) {
            			readProcess(irList, line);
            		} else if (line.startsWith("delay")) {
            			setDelay(line);
            		} else if (line.startsWith("period")) {
            			setPeriod(line);
            		} else if (line.startsWith("maxWait")) {
            			setWait(line); 
            		}
            	}
            }
            
            //Change made for Lab1
            Collections.sort(nodeNames);
            
            br.close();
            fr.close();

            
		} catch (FileNotFoundException e) {
			e.printStackTrace();
            return;
        } catch (IOException e) {
        	try { br.close(); } catch (IOException e2) { /* Ignored */ }
            return;
        }   
	}
	
	
	/**
	 * Re-read the configuration file and refresh the configuration of the system.
	 * ME, NAMES and node.<name>.ip/port are referenced during initial setup up
	 * and are not refreshed in this method.
	 * 
	 */
	public void refreshConfig() {
		csList.clear();
		crList.clear();
		isList.clear();
		irList.clear();
		
		BufferedReader br = null;
		try {
			FileReader fr = new FileReader(configFile);
            br = new BufferedReader(fr);
            
            boolean eof = false;
            while (!eof) {
            	String line = br.readLine();
            	if (line == null) {
                    eof = true;
            	} else {
            		if (line.startsWith("cs")) {
            			readProcess(csList, line);
            		} else if (line.startsWith("cr")) {
            			readProcess(crList, line);
            		} else if (line.startsWith("is")) {
            			readProcess(isList, line);
            		} else if (line.startsWith("ir")) {
            			readProcess(irList, line);
            		} 
            	}
            }
            
            br.close();
            fr.close();
            
		} catch (FileNotFoundException e) {
			e.printStackTrace();
            return;
        } catch (IOException e) {
        	try { br.close(); } catch (IOException e2) { /* Ignored */ }
            return;
        }   
	}
	
	/**
	 * Read a line from the configuration file starting with keyword "ME".
	 * parse and get the name of the node if applicable to configure the MessagePasser
	 * 
	 * @param line a line in the configuration file
	 * @see #readConfig(String)
	 */
	public void readMyName(String line) {
		String cut = line.substring(line.indexOf("ME") + 2);
		myName = cut.trim();
	}
	
	/**
	 * Read a line from the configuration file starting with keyword "NAMES".
	 * parse and get the names of the nodes in the system to configure the MessagePasser
	 * 
	 * @param line a line in the configuration file
	 * @see #readConfig(String)
	 */
	public void readNames(String line) {
		String cut = line.substring(line.indexOf("NAMES") + 6);
		String names[] = cut.split(",");
		for(String name : names) {
			//Change made for LAB1
			nodeNames.add(name);
		}
	}
	
	/**
	 * Read a line from the configuration file starting with keyword "nodes.".
	 * parse and get the ip/ports of the nodes in the system to configure the MessagePasser
	 * 
	 * @param line a line in the configuration file
	 * @see #readConfig(String)
	 */
	public void readNodes(String line) {
		String name = find2ndWord(line);
		if(line.contains(".ip")) {
			String cut = line.substring(line.indexOf(".ip") + 3);
			String ip = cut.trim();
			NodeInfo info = new NodeInfo(ip);
			nodeList.put(name, info);
		}
		
		if(line.contains(".port")) {
			String cut = line.substring(line.indexOf(".port") + 5);
			String portRaw = cut.trim();
			int p = Integer.parseInt(portRaw);
			NodeInfo info = nodeList.get(name);
			
		//	System.out.println(name + " " + p);
			info.setPort(p);
			nodeList.put(name, info);
		}
	}
	
	/**
	 * Read a line from the configuration file.
	 * parse and store the process info of certain message into a map.
	 * Later, MessagePasser will depend on such a map to decide what to do with a message
	 * 
	 * @param map a map storing information about how a message with certain id or of certain class should be processed
	 * @param line a line in the configuration file
	 * @see #readConfig(String)
	 */
	public void readProcess(Map<String, String> map, String line) {
		String name = find2ndWord(line);
		if (line.contains(" drop")) {
			map.put(name, "drop");
		} else if (line.contains(" delay")) {
			map.put(name, "delay");
		} else if (line.contains(" duplicate")) {
			map.put(name, "duplicate");
		}
	}
	
	/**
	 * A helper method for parsing the configuration file.
	 * Find the second word in a pattern of ***.SecondWord.***
	 * 
	 * @param line a line in the configuration file
	 * @return the second word in the line
	 * @see #readNodes(String)
	 * @see #readProcess(Map, String)
	 */
	public String find2ndWord(String line) {
		String cut1 = line.substring( line.indexOf('.') + 1);
		String cut2;
		if (cut1.contains(".")) {
			cut2 = cut1.substring(0, cut1.indexOf('.'));
		} else {
			String splits[] = cut1.split(" ");
			cut2 = splits[0];
		}
		
		return cut2;
	}
	
	/**
	 * Send a message.
	 * <p>
	 * The actual action taken will depend on the header of the message
	 * 
	 * @param message
	 * @see #readProcess(Map, String)
	 * @see #sendDecision(Message)
	 */
	@SuppressWarnings("unchecked")
	public String send(Message message) {

		refreshConfig();
				
		String[] destination;
		if (message.isLockMsg()) {
			destination = (String[]) nodeNames.toArray(new String[0]);
		} else {
			destination = message.getMultiDestination();
		}
		
		//XXX
		if (message.isLockMsg() == false) {
			message.msgType = new String ("data");
			
		}
		
		for(String dest: destination) {
			if (!nodeNames.contains(dest)) {
				return "destination does not exist";
			}
		}
		this.sendStamp((TimeStampedMessage)message);
		
		if (message.isMulticastMsg()) {
			
			((TimeStampedMessage)message).setSequenceNumber((Map<String, Integer>)((HashMap<String, Integer>)this.sequenceNumber).clone());
			((TimeStampedMessage)message).setOrgSender(this.myName);
			this.seqNo++;
			
			
			((TimeStampedMessage)message).setSeqNo(this.seqNo);
			
				((TimeStampedMessage)message).initAckList(destination);
				senderQ.add((TimeStampedMessage)message);
			
		}
		
		for (String dest: destination) {
			TimeStampedMessage msg = ((TimeStampedMessage)message).serializeFor(dest);
			sendDecision(msg);
		}
		
		return "send success";
	}
	
	public List<TimeStampedMessage> getHbQ() {
		return hbQ;
	}

	public void setHbQ(List<TimeStampedMessage> hbQ) {
		this.hbQ = hbQ;
	}

	public Map<String, Integer> getSequenceNumber() {
		return sequenceNumber;
	}

	public void setSequenceNumber(Map<String, Integer> sequenceNumber) {
		this.sequenceNumber = sequenceNumber;
	}

	/**
	 * Decide what to do with a message based on its ID or msg_class.
	 * A message may be sent directly, may be dropped,
	 * may be delayed and put into output buffer, or duplicated
	 * 
	 * @param msg
	 */
	public void sendDecision(Message msg) {
		Header header = msg.getHeader();
		String is = header.getID();
		String cs = header.getMsgClass();
		
		String operation = "";
		
		if(isList.containsKey(is)) {
			operation = isList.get(is);
		} else if (csList.containsKey(cs)) {
			operation = csList.get(cs);
		}
		
		//changes made here
		
		if (operation.equals("drop")) {
			return;
		} else if (operation.equals("delay")) {
			outputBuffer.add(msg);
			return;
		} else if (operation.equals("duplicate")) {
			outputBuffer.add(msg);
		}
		
		socketSend(msg);
		while(outputBuffer.size() > 0) {
			socketSend(outputBuffer.get(0));
			outputBuffer.remove(0);
		}
		
	}
	
	/**
	 * Connect the socket to the destination as is indicated in the messages header 
	 * and send the message
	 * 
	 * @param msg
	 * @see #sendDecision(Message)
	 */
	public void socketSend(Message msg){
		
		
		msg.setPhySender(myName);
		Socket myClient;
		String clientName = msg.getHeader().getDest();
		System.out.println ("Sending " + msg.getMsgType() + "to " + clientName);
				
		try {
			myClient = new Socket(nodeList.get(clientName).getIp(),nodeList.get(clientName).getPort());
			
			ObjectOutputStream output;
		    output = new ObjectOutputStream(myClient.getOutputStream());
		    output.writeObject(msg);
		    output.close();
		}
		catch (IOException e) {
			System.out.println("Host " + clientName + " unknown. Message send fail");
		}
		catch (Exception e) {
			System.out.println(" exception Host " + clientName + " unknown. Message send fail");
		}
	}
	
	/**
	 * Receive a message.
	 * <p>
	 * Accept a message in the inputBuffer. 
	 * The actual action taken is dependent on the message id, class, and process configuration
	 * 
	 * If the message is delayed or duplicated, 
	 * we put the delayed msg or the duplicated copy into the ddBuffer.
	 * And every time we call receive(), we check the ddBuffer first. 
	 * If ddBuffer is not empty, we deliver the msg in ddBuffer first. 
	 * if it's empty, then we go to inputBuffer to deliver new message
	 * 
	 */
	@SuppressWarnings("unchecked")
	public Message receive() {
	
		//comment out for lab3
		Collections.sort(this.inputBuffer);
		refreshConfig();
		Message msg;
		if (this.inputBuffer.size() == 0) {
			return null;
		}
		Message m = this.inputBuffer.get(0);
		inputBuffer.remove(0);
		return m;
	}

	/**
	 * Find out how a message should be processed when RECEIVED
	 * 
	 * @param message
	 * @return The operation to be taken with the message, 
	 * which may be receive directly, drop, delay or duplicate
	 * 
	 */
	public String findOp(Message message) {
		String ir = message.getHeader().getID();
		String cr = message.getHeader().getMsgClass();
		
		String operation = "";
		
		if(irList.containsKey(ir)) {
			operation = irList.get(ir);
		} else if (crList.containsKey(cr)) {
			operation = crList.get(cr);
		}
		return operation;
	}
	
	
	public List<Message> getInputBuffer() {
		return inputBuffer;
	}

	public void setInputBuffer(List<Message> inputBuffer) {
		this.inputBuffer = inputBuffer;
	}

	public List<Message> getOutputBuffer() {
		return outputBuffer;
	}

	public void setOutputBuffer(List<Message> outputBuffer) {
		this.outputBuffer = outputBuffer;
	}
	
	public String getMyName() {
		return this.myName;
	}

	////////////////////////LAB1/////////////////////////////////
	public void sendStamp(TimeStampedMessage msg) {
		clock.doService();
		msg.attatchStamp((ClockService)(clock.clone()));
		msg.setSenderStamp((ClockService)(clock.clone())); // lab2
	}
	
	////////////////V9 change/////////////////
	public void recvStamp(TimeStampedMessage msg) {
		//original stamp (sender's)
		ClockService senderStamp = msg.getStamp();
		
		//attatch new stamp (receiver's)
		clock.syncClock(nodeNames,msg.getStamp());
		clock.doService();
		msg.attatchStamp((ClockService)(clock.clone()));
		
	}
		
	public List<String> getNodeNames() {
		return nodeNames;
	}
	
	
	public List<TimeStampedMessage> getSenderQ() {
		return senderQ;
	}

	public void setSenderQ(List<TimeStampedMessage> senderQ) {
		this.senderQ = senderQ;
	}
	
	public long getDelay() {
		return delay;
	}
	
	public long getPeriod() {
		return period;
	}

	/**
	 * if a message is multicast to every node successfully
	 * clear it from the the sender queue storing the maybe-success sent msgs
	 * 
	 * @param message
	 */
	public void clearFromSenderQ(TimeStampedMessage message) {
		for (int index = 0; index < senderQ.size(); index++) {
			if (senderQ.get(index).getSenderStamp() == message.getSenderStamp()) {
				senderQ.remove(index);
				break;
			}
		}
	}
	
	
	
	public void addToNetworkBuffer(Message message) {
		nwBuffer.add(message);
	}
	
	public void addToReadyBuffer(Message message) {
		readyBuffer.add(message);
	}
	
	public void networkToReady() {
		for (Message message : nwBuffer) {
			readyBuffer.add(message);
		}
		nwBuffer.clear();
	}
	
	public boolean readyToReceive() {
		return !readyBuffer.isEmpty();
	}
	
	public List<Message> getReadyBuffer() {
		return readyBuffer;
	}
	
	public void emptyReadyBuffer() {
		readyBuffer.clear();
	}
	

	
	// update the ACK list of the message in the hbQ
	void updateAckOfMsg(TimeStampedMessage msg){
		TimeStampedMessage message = null;
		int index = 0;
		boolean allDone = true;
		System.out.println("Ack " + msg.getHeader().getSrc() + " Seq" + msg.getSeqNo());
				
		for (index = 0; index < senderQ.size(); index++) {
			if (senderQ.get(index).getOrgSender().equals(msg.getOrgSender()) && msg.getSeqNo() == (senderQ.get(index).getSeqNo())) {
		//		System.out.println("returned sender message");
				message = senderQ.get(index);
				message.getAckList().put(msg.getHeader().getSrc(), new Boolean("true"));
				for (String receiver : message.ackList.keySet()) {
					if (message.ackList.get(receiver) == false) {
						allDone = false;
						break;
					}
				}
				if (allDone == true) {
					sendOkMessage(message);
					senderQ.remove(message);
					break;
				}
			}
		}
		
		
	}
	
	boolean redundantMessage(TimeStampedMessage msg){
		// already in hbQ
		for (TimeStampedMessage m : hbQ) {
			
			if(m.getOrgSender().contains(msg.getOrgSender()) && (m.getSeqNo() == msg.getSeqNo())) {
				System.out.println("true");
				return true;
			}
		}
		
		// smaller than expected, meaning that its outdated data
		if (sequenceNumber.get(msg.getOrgSender()).intValue() >= msg.getSeqNo()) {
			System.out.println(" seq true");
			return true;
		}
		
		return false;
		
		
	}
	
	public void acceptRealMessage(TimeStampedMessage mesg) {

		// no matter redundant or not, do the following
		System.out.println("accepting real message %%%%%%%%");
			
		
		//if it is a new message, add it to hbQ
		if (redundantMessage(mesg) == true) {
			System.out.println("redundant " + mesg.getPayload());
		}
		
		if (redundantMessage(mesg) == false) {
			
			String senderName = mesg.getPhySender();
			String senderTime = ((Vector)mesg.getSenderStamp()).getTime();
			receivedMsg.add(senderName + senderTime);
			
			hbQ.add(mesg);
			
			sendAckMessage(mesg);
			
			// start the receive check timer
			
			
		}
	}
	
	void sendAckMessage(TimeStampedMessage msg) {

		String node = msg.physicalSender;
	//	System.out.println("sending ack to *********" + node);
		TimeStampedMessage message = new TimeStampedMessage (node, myName, "default", "default",
															 null, msg.isloggable, msg.isMulticastMsg()); 
		message.setMsgType("ack");
		message.attatchStamp((Vector)msg.getStamp().clone());
		message.setSenderStamp((Vector)msg.getSenderStamp().clone());
		message.setSequenceNumber(sequenceNumber);
		message.setSeqNo(msg.getSeqNo());
		message.setOrgSender(msg.getOrgSender());
		socketSend((Message)message);
	}
	
	public boolean isExpecting(TimeStampedMessage message) {
		String origin = message.getOrgSender();
		int expecting = sequenceNumber.get(origin) + 1;
	//	System.out.println("expect check seq" + message.getSeqNo() + " " + message.getPayload());
	//	System.out.println("expecting " + expecting);
		if (message.getSeqNo() == expecting) {
			return true;
		} else {
			return false;
		}
		
	}
	
	public boolean isNextToExpecting(TimeStampedMessage message) {
		String origin = message.getOrgSender();
		int nextExpecting = sequenceNumber.get(origin) + 2;
	//	System.out.println("expect check seq" + message.getSeqNo() + " " + message.getPayload());
	//	System.out.println("expecting " + nextExpecting);
		if (message.getSeqNo() == nextExpecting) {
			return true;
		} else {
			return false;
		}
		
	}
	
	public boolean isOutDated(TimeStampedMessage message) {
		String origin = message.getOrgSender();
		int expecting = sequenceNumber.get(origin) + 1;
	//	System.out.println("outdate check seq" + message.getSeqNo() + " " + message.getPayload());
	//	System.out.println("curr expecting " + expecting);
		if (message.getSeqNo() < expecting) {
			return true;
		} else {
			return false;
		}
		
	}
	
	
	
	public void clearOutdatedHbQ(TimeStampedMessage message) {
		List <TimeStampedMessage> toBeCleared = new ArrayList<TimeStampedMessage>();
		for (TimeStampedMessage m : hbQ) {
			if (m.getOrgSender().contains(message.getOrgSender()) && m.getSeqNo() < message.getSeqNo()) {
				toBeCleared.add(m);
			}
		}
		for (TimeStampedMessage m : toBeCleared) {
			hbQ.remove(m);
		}
	}
	
	public void setDelay(String line) {
		String d = line.split(" ")[1];
		delay = Long.parseLong(d.trim());
	}
	
	public void setPeriod(String line) {
		String p = line.split(" ")[1];
		period = Long.parseLong(p.trim());
	}
	
	public void setWait(String line) {
		String w = line.split(" ")[1];
		wait = Integer.parseInt(w.trim());
	}
	
	public int getWait() {
		return wait;
	}

	public void sendOkMessage(TimeStampedMessage msg) {
		
			for (String node : nodeNames) {
			
				TimeStampedMessage message = new TimeStampedMessage (node, myName, "default", "default",
																	 null, msg.isloggable, msg.isMulticastMsg()); 
				message.setMsgType("ok");
				message.attatchStamp((Vector)msg.getStamp().clone());
				message.setSenderStamp((Vector)msg.getSenderStamp().clone());
				message.setSequenceNumber(sequenceNumber);
				message.setSeqNo(msg.getSeqNo());
				message.setOrgSender(msg.getOrgSender());
				socketSend((Message)message);
			}
	}
	
	public void moveToInputBuffer(TimeStampedMessage msg) {
		
		TimeStampedMessage message = null;
		int index = 0;
		
		for (index = 0; index < hbQ.size(); index++) {
			if (hbQ.get(index).getOrgSender().equals(msg.getOrgSender()) && msg.getSeqNo() == (hbQ.get(index).getSeqNo())) {
				System.out.println("returned sender message");
				this.inputBuffer.add(hbQ.remove(index));
				break;
			}
		}
	}
}



